package com.ada.spark.rddoperator

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 作用：根据分区数，重新通过网络随机洗牌所有数据。
  */
object Spark12_repartition {

    def main(args: Array[String]): Unit = {
        //创建SparkConf
        val conf = new SparkConf().setAppName("Spark12_repartition").setMaster("local[*]")
        //创建Spark上下文对象
        val sc = new SparkContext(conf)

        //缩减分区数：可以简单理解为合并分区数
        val listRDD: RDD[Int] = sc.makeRDD(1 to 16, 4)

        println("缩减分区前：" + listRDD.partitions.size)

        //repartition实际上是调用的coalesce，默认是进行shuffle的, 源码如下：
        /*
        def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
            coalesce(numPartitions, shuffle = true)
        }
         */
        val coalesceRDD: RDD[Int] = listRDD.repartition(2)

        println("缩减分区后：" + coalesceRDD.partitions.size)

        coalesceRDD.saveAsTextFile("output")
    }

    /**
      * coalesce和repartition的区别
      * 1、coalesce重新分区，可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定。
      * 2、repartition实际上是调用的coalesce，默认是进行shuffle的。源码如下：
      * def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
      * coalesce(numPartitions, shuffle = true)
      * }
      */

}
